package com.shujia.core

import org.apache.flink.streaming.api.scala._

object Demo1WordCount {
  def main(args: Array[String]): Unit = {
    /**
      *
      * 1、创建flink流处理的环境
      *
      */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //设置flink的并行度
    env.setParallelism(1)

    //上游task将数据发生到下游task的最大延迟时间
    //如果再200毫秒内数据达到了332k，也会发生
    //如果设置为0,每一条数据发送一次，吞吐量会降低（每一秒处理的数据量）
    env.setBufferTimeout(200)

    /**
      * 2、读取数据
      *
      * 读取socket的数据
      * nc -lk 8888
      *
      */

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
      *
      * 3、将一行转换成多行
      */

    val wordsDS: DataStream[String] = linesDS.flatMap(line => line.split(","))

    /**
      * 4、转换成KV格式
      *
      */

    val kvDS: DataStream[(String, Int)] = wordsDS.map(word => (word, 1))


    /**
      * 5、按照单词分组
      *
      */
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(kv => kv._1)


    /**
      * 6、对value求和
      *
      */
    //通过下标指定对哪一个列求和
    val countDS: DataStream[(String, Int)] = keyByDS.sum(1)


    //打印结果
    countDS.print()

    /**
      * 启动flink程序
      *
      */
    env.execute()
  }
}
